# 导入库
import pandas as pd
import numpy as np
import os
from sqlalchemy import create_engine

# ------------------- 1. 基础配置（脱敏处理，占位符替换）-------------------
db_config = {
    "host": "【数据库地址占位符】",
    "user": "【数据库用户名占位符】",
    "password": "******",
    "database": "lendingclub_loan_db_v1",
    "port": 3306,
    "table_name": "loan_2018_q3"
}
# 结果保存路径（占位符）
result_dir = "【结果保存路径占位符】"
os.makedirs(result_dir, exist_ok=True)

# ------------------- 2. 核心测试函数（保留规则验证核心逻辑）-------------------
def load_data(engine):
    """读取规则验证所需核心字段，字段存在性校验"""
    needed_fields = [
        "installment", "annual_inc", "loan_status", "dti", 
        "delinq_2yrs", "inq_last_6mths", "open_acc", "revol_util",
        "emp_length", "home_ownership", "fico_range_low"
    ]
    try:
        # 字段存在性校验
        table_cols = pd.read_sql_query(f"DESCRIBE {db_config['table_name']}", con=engine)["Field"].tolist()
        missing_fields = [f for f in needed_fields if f not in table_cols]
        if missing_fields:
            print(f"❌ 表中缺失规则必需字段：{missing_fields}")
            exit()
        
        df = pd.read_sql_table(db_config["table_name"], con=engine, columns=needed_fields)
        print(f"✅ 数据读取成功：{df.shape[0]}行×{df.shape[1]}列")
        return df
    except Exception as e:
        print(f"❌ 数据读取失败：{str(e)}")
        exit()

def preprocess_data(df):
    """数据预处理：适配规则验证的格式转换与异常值过滤"""
    # 字段类型转换
    df["emp_length"] = pd.to_numeric(df["emp_length"], errors="coerce")
    
    # 异常值过滤
    df_clean = df[
        (df["annual_inc"] > 0) & (df["annual_inc"] <= 1000000)
        & (df["dti"].between(0, 100))
        & (df["revol_util"].between(0, 100))
        & (df["open_acc"].between(0, 20))
        & (df["delinq_2yrs"] >= 0)
        & (df["inq_last_6mths"] >= 0)
        & (df["loan_status"].isin([0, 1]))
        & (df["emp_length"].notna())
        & (df["home_ownership"].isin(["RENT", "OWN", "MORTGAGE"]))
    ].copy()
    
    # 计算核心特征：月还款压力
    df_clean["monthly_pressure"] = df_clean["installment"] / (df_clean["annual_inc"] / 12)
    df_clean = df_clean[df_clean["monthly_pressure"] <= 1]
    
    # 缺失值处理
    rule_necessary_fields = [
        "monthly_pressure", "revol_util", "delinq_2yrs", "inq_last_6mths", 
        "open_acc", "dti", "emp_length", "home_ownership", "fico_range_low"
    ]
    df_clean = df_clean.dropna(subset=rule_necessary_fields)
    
    print(f"✅ 数据预处理完成：{df_clean.shape[0]}行有效数据")
    return df_clean

def apply_all_rules(df):
    """应用风控规则：硬规则拒贷→低风险通过→审核关注"""
    # 硬规则（命中即拒贷）
    hard_rule1 = (df["monthly_pressure"] >= 0.25) & (df["inq_last_6mths"] >= 3) & (df["emp_length"] < 2)
    hard_rule2 = (df["open_acc"].between(0, 4)) & (df["revol_util"].between(0, 10)) & (df["dti"].between(20, 36))
    hard_rule3 = (df["home_ownership"].isin(["RENT", "OWN"])) & (df["monthly_pressure"] >= 0.25) & (df["emp_length"] < 1)
    hard_rule4 = (df["monthly_pressure"] >= 0.25) & (df["delinq_2yrs"] >= 3)
    
    # 计算单条硬规则误拒率
    def calc_single_hard_rule_misreject(rule, df):
        hit_df = df[rule]
        if len(hit_df) == 0:
            return 0.0
        misreject_count = (hit_df["loan_status"] == 0).sum()
        return misreject_count / len(hit_df) * 100
    
    # 硬规则统计
    hard_rule_misrejects = [
        calc_single_hard_rule_misreject(hard_rule1, df),
        calc_single_hard_rule_misreject(hard_rule2, df),
        calc_single_hard_rule_misreject(hard_rule3, df),
        calc_single_hard_rule_misreject(hard_rule4, df)
    ]
    hard_rules = [("硬规则1", hard_rule1, hard_rule_misrejects[0]), 
                  ("硬规则2", hard_rule2, hard_rule_misrejects[1]),
                  ("硬规则3", hard_rule3, hard_rule_misrejects[2]),
                  ("硬规则4", hard_rule4, hard_rule_misrejects[3])]
    
    print(f"\n📊 硬规则命中详情（含误拒率）：")
    for name, rule, misreject in hard_rules:
        count = rule.sum()
        ratio = count / len(df) * 100
        print(f"  {name}：命中{count}条（占比{ratio:.2f}%）| 误拒率：{misreject:.2f}%")
    
    # 合并硬规则
    is_high_risk = hard_rule1 | hard_rule2 | hard_rule3 | hard_rule4
    
    # 低风险规则（命中即通过）
    low_risk1 = (df["fico_range_low"].between(750, 850)) & (df["delinq_2yrs"] == 0)
    low_risk2 = (df["home_ownership"] == "MORTGAGE") & (df["emp_length"] >= 10) & (df["dti"].between(10, 20))
    low_risk3 = (df["open_acc"] >= 12) & (df["revol_util"].between(0, 10)) & (df["dti"].between(10, 20))
    low_risk4 = (df["monthly_pressure"].between(0, 0.1)) & (df["inq_last_6mths"] == 0) & (df["delinq_2yrs"] == 0)
    
    # 合并低风险规则
    is_low_risk = ~is_high_risk & (low_risk1 | low_risk2 | low_risk3 | low_risk4)
    
    # 审核关注用户
    is_review = ~is_high_risk & ~is_low_risk
    
    # 标记风险等级
    df["risk_level"] = "未知"
    df.loc[is_high_risk, "risk_level"] = "硬规则拒贷"
    df.loc[is_low_risk, "risk_level"] = "低风险通过"
    df.loc[is_review, "risk_level"] = "审核关注"
    
    # 低风险规则统计
    print(f"\n📊 低风险规则命中详情：")
    low_rules = [("低风险1", low_risk1), ("低风险2", low_risk2), ("低风险3", low_risk3), ("低风险4", low_risk4)]
    for name, rule in low_rules:
        valid_count = rule.sum()
        ratio = valid_count / len(df) * 100
        print(f"  {name}：命中{valid_count}条（占比{ratio:.2f}%）")
    
    # 最终结果拆分
    high_risk_df = df[is_high_risk].copy()
    low_risk_df = df[is_low_risk].copy()
    review_df = df[is_review].copy()
    
    print(f"\n🎯 最终筛选结果：")
    print(f"  硬规则拒贷：{len(high_risk_df)}条（占比{len(high_risk_df)/len(df)*100:.2f}%）")
    print(f"  低风险通过：{len(low_risk_df)}条（占比{len(low_risk_df)/len(df)*100:.2f}%）")
    print(f"  审核关注：{len(review_df)}条（占比{len(review_df)/len(df)*100:.2f}%）")
    
    return low_risk_df, review_df, high_risk_df

def calculate_core_metrics(df, low_risk, review, high_risk):
    """计算风控核心指标，验证规则有效性"""
    total = len(df)
    high_risk_cnt = len(high_risk)
    low_risk_cnt = len(low_risk)
    review_cnt = len(review)
    
    # 核心指标计算
    metrics = {
        "总有效客群数": total,
        "硬规则拒贷数": high_risk_cnt,
        "低风险通过数": low_risk_cnt,
        "审核关注数": review_cnt,
        "硬规则拒贷率(%)": round(high_risk_cnt/total*100, 2) if total>0 else 0,
        "低风险通过率(%)": round(low_risk_cnt/total*100, 2) if total>0 else 0,
        "审核关注率(%)": round(review_cnt/total*100, 2) if total>0 else 0,
        "原始逾期率(%)": round(df["loan_status"].mean()*100, 2) if total>0 else 0,
        "硬规则拒贷用户逾期率(%)": round(high_risk["loan_status"].mean()*100, 2) if high_risk_cnt>0 else 0,
        "低风险通过用户逾期率(%)": round(low_risk["loan_status"].mean()*100, 2) if low_risk_cnt>0 else 0,
        "审核关注用户逾期率(%)": round(review["loan_status"].mean()*100, 2) if review_cnt>0 else 0,
        "最终实际通过率(%)": round((low_risk_cnt + review_cnt)/total*100, 2) if total>0 else 0,
        "误拒率(%)": round((high_risk["loan_status"]==0).sum()/high_risk_cnt*100, 2) if high_risk_cnt>0 else 0
    }
    
    # 打印指标汇总
    print(f"\n📈 风控核心指标汇总：")
    for key, value in metrics.items():
        print(f"  {key}：{value}")
    
    # 规则有效性判断
    if metrics["原始逾期率(%)"] > metrics["低风险通过用户逾期率(%)"]:
        print(f"\n✅ 规则生效：低风险用户逾期率低于原始逾期率")
    if metrics["硬规则拒贷用户逾期率(%)"] > metrics["原始逾期率(%)"]:
        print(f"✅ 规则精准：硬规则有效拦截高风险用户")
    if metrics["误拒率(%)"] > 30:
        print(f"⚠️  注意：误拒率过高，建议优化硬规则")
    
    return metrics

def save_results(metrics, low_risk, review, high_risk):
    """保存规则验证结果到Excel"""
    excel_path = os.path.join(result_dir, "风控规则验证结果.xlsx")
    try:
        with pd.ExcelWriter(excel_path, engine="openpyxl") as writer:
            # 核心指标sheet
            pd.DataFrame([metrics]).to_excel(writer, sheet_name="核心指标汇总", index=False)
            # 各风险等级用户详情
            detail_fields = ["monthly_pressure", "revol_util", "delinq_2yrs", "inq_last_6mths", 
                            "open_acc", "dti", "emp_length", "home_ownership", "fico_range_low", 
                            "loan_status", "risk_level"]
            low_risk[detail_fields].to_excel(writer, sheet_name="低风险通过用户", index=False)
            review[detail_fields].to_excel(writer, sheet_name="审核关注用户", index=False)
            high_risk[detail_fields].to_excel(writer, sheet_name="硬规则拒贷用户", index=False)
        
        print(f"\n✅ 结果保存成功：{excel_path}")
    except Exception as e:
        print(f"❌ 结果保存失败：{str(e)}")

# ------------------- 3. 主测试逻辑 -------------------
if __name__ == "__main__":
    print("="*60)
    print("=== 风控规则验证程序 ===")
    print("="*60)
    
    # 建立数据库连接
    try:
        engine = create_engine(
            f"mysql+pymysql://{db_config['user']}:{db_config['password']}@{db_config['host']}:{db_config['port']}/{db_config['database']}"
        )
        print("✅ 数据库连接成功")
    except Exception as e:
        print(f"❌ 数据库连接失败：{str(e)}")
        exit()
    
    # 执行测试流程
    df_raw = load_data(engine)
    df_clean = preprocess_data(df_raw)
    low_risk_df, review_df, high_risk_df = apply_all_rules(df_clean)
    metrics = calculate_core_metrics(df_clean, low_risk_df, review_df, high_risk_df)
    save_results(metrics, low_risk_df, review_df, high_risk_df)
    
    print("\n" + "="*60)
    print("=== 规则验证程序执行完成 ===")
    print("="*60)

